/*
 * The MIT License
 *
 * Copyright 2014 David Rodenas Pico.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

package edu.uoc.sd.fr14.comp;

import java.util.Objects;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import edu.uoc.sd.fr14.actors.Clock;
import edu.uoc.sd.fr14.ClockConfiguration;
import edu.uoc.sd.fr14.mesg.Message;
import edu.uoc.sd.fr14.mesg.TimeoutElectionBeginMessage;
import edu.uoc.sd.fr14.mesg.TimeoutElectionEndMessage;
import edu.uoc.sd.fr14.mesg.TimeoutMessage;
import edu.uoc.sd.fr14.mesg.TimeoutSynchronizeMessage;
import edu.uoc.sd.fr14.model.CustomerId;
import edu.uoc.sd.fr14.model.FlightId;
import edu.uoc.sd.fr14.model.Logger;
import edu.uoc.sd.fr14.model.OpId;
import edu.uoc.sd.fr14.model.ServerId;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * This is a component that is a clock for servers.
 * It is prepared to be called directly to generate later locally cancelable clock messages.
 * 
 * Output:
 * - output: output of the clock messages.
 * 
 * It starts a clock task that never is canceled,
 * instead of this it registers all messages sent or waiting to send 
 * and cancels them. 
 * The idea is from one single thread, a canceled message in a previous one,
 * will never be executed.
 * 
 * @author drodenas
 */
public class ClockComponent extends Component implements Clock {
    
    private static final Logger log = Logger.get("ClockComponent");
    
    private final Set<ServerId> localServers;
    private final ClockConfiguration serverConfiguration;
    private final Object messagesGuard = new Object();
    private final Map<Object,List<TimeoutMessage>> messages = new ConcurrentHashMap<>();
    private final Map<ServerId,List<Object>> aliveKeys = new ConcurrentHashMap<>();

    public ClockComponent(Set<ServerId> localServers, ClockConfiguration serverConfiguration) {
        super(localServers+"-clock");
        this.localServers = localServers;
        this.serverConfiguration = serverConfiguration;
        
        // others
        scheduleSynchronize();
    }

    @Override
    public void cancelPrepareTo(OpId op, ServerId from) {
        cancel(new PrepareKey(op, from));
    }

    @Override
    public void cancelSeatTo(FlightId flight, CustomerId customer, ServerId from) {
        cancel(new SeatKey(flight, customer, from));
    }

    @Override
    public void setElectionTo(ServerId from) {
        set(from, from);
    }

    @Override
    public void setPrepareTo(OpId op, ServerId from) {
        set(new PrepareKey(op, from), from);
    }

    @Override
    public void setSeatTo(FlightId flight, CustomerId customer, ServerId from) {
        set(new SeatKey(flight, customer, from), from);
    }

    @Override
    protected void process(Message message) {
    }
    
    private void cancel(Object key) {
        synchronized (messagesGuard) {
            List<TimeoutMessage> alives = messages.get(key);
            
            if (alives == null) {
                log.debug("X", "cancelTo "+key+" with no message");
                return;
            }
            
            // cancel the first, the oldest the first
            TimeoutMessage to = alives.remove(0);
            aliveKeys.get(to.getTarget()).remove(key);
            log.debug(to.getTarget(), "cancelTo "+key+" message '"+to+"'");
            to.cancel();
            if (alives.isEmpty()) {
                messages.remove(key);
            }
        }
    }

	private void cancelAllFrom(ServerId target) {
        log.debug(target, "cancelAllFrom");
        
		List<Object> keys = new ArrayList<>();
		synchronized (messagesGuard) {
			// if no exists do nothing
			if (!aliveKeys.containsKey(target)) {
				return;
			}
			
			// remove all
			keys.addAll(aliveKeys.get(target));
			for (Object key : keys) {
				cancel(key);
			}
			aliveKeys.get(target).clear();			
		}
	}
    
    
    private void set(final Object key, final ServerId target) {
    	// before start, definition of the message and their trigger of done
        final TimeoutMessage message = new TimeoutElectionBeginMessage(target, new TimeoutMessage.Finally() {
            @Override
            public void finish(TimeoutMessage me, boolean success) {
            	// when done remove the message from the queue
                synchronized (messagesGuard) {
                    List<TimeoutMessage> alives = messages.get(key);
                    if (alives != null) {
                        alives.remove(me);
                        if ((aliveKeys!=null) && (target!=null) && (key!=null)){ 
                            if (aliveKeys.containsKey(target)){ 
                               if (aliveKeys.get(target).contains(key)){ 
                                     aliveKeys.get(target).remove(key); 
                                         if (alives.isEmpty()) { 
                                             messages.remove(key); 
                                             aliveKeys.remove(target); 
                                         } 
                                     } 
                                 } 
                             }
                    }
                }
                
                // if success cancel others and trigger election end
                if (success) {
                	cancelAllFrom(target);
                    scheduleElectionEnd(target);
                }
            }
        });
        
        // set the timeout for this server
        log.debug(target, "setTo "+key+" message '"+message+"'");
        synchronized (messagesGuard) {
            List<TimeoutMessage> alives = messages.get(key);
            List<Object> keys = aliveKeys.get(target);
            if (alives == null) {
                alives = new LinkedList<>();
                keys = new LinkedList<>();
                messages.put(key, alives);
                aliveKeys.put(target, keys);
            }
            // add to the end, the newest is the last
            alives.add(message);
            keys.add(key);
        }
        schedule(message, target);
    }
    
    private void schedule(final TimeoutMessage message, final ServerId target) {
        getTimer().schedule(new TimerTask() {
           @Override
            public void run() {
                deliver(message);
            }
        }, serverConfiguration.getElectionBeginTo());
    }

    private void scheduleElectionEnd(final ServerId target) {
        getTimer().schedule(new TimerTask() {
           @Override
            public void run() {
                TimeoutMessage message = new TimeoutElectionEndMessage(target,TimeoutMessage.NOTHING);
                deliver(message);
            }
        }, serverConfiguration.getElectionEndTo());
    }

    private TimerTask scheduleSynchronize() {
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                synchronize();
            }
        };
        getTimer().schedule(task, serverConfiguration.getSynchronizeTo(), serverConfiguration.getSynchronizeTo());
        return task;
    }
    
    private void synchronize() {
        for (ServerId target : localServers) {
            Message message = new TimeoutSynchronizeMessage(target);
            deliver(message);
        }
    }

    @Override
    protected void deliver(Message message) {
        log.debug(message.getTarget(), "deliver '"+message+"'");
        super.deliver(message);
    }
    
    @Override
    public void stop() {
        super.stop();
        
        List<TimeoutMessage> cancels = new LinkedList<>();
        synchronized (messagesGuard) {
            for (List<TimeoutMessage> alives : messages.values()) {
                for (TimeoutMessage message : alives) {
                    cancels.add(message);
                }
            }
            messages.clear();
        }
        
        for (TimeoutMessage message : cancels) {
            message.cancel();
        }
    }
    
    private static class PrepareKey {
        private final OpId op;
        private final ServerId from;

        public PrepareKey(OpId op, ServerId from) {
            this.op = op;
            this.from = from;
        }

        @Override
        public int hashCode() {
            int hash = 7;
            hash = 43 * hash + Objects.hashCode(this.op);
            hash = 43 * hash + Objects.hashCode(this.from);
            return hash;
        }

        @Override
        public boolean equals(Object obj) {
            if (obj == null) {
                return false;
            }
            if (getClass() != obj.getClass()) {
                return false;
            }
            final PrepareKey other = (PrepareKey) obj;
            if (!Objects.equals(this.op, other.op)) {
                return false;
            }
            return Objects.equals(this.from, other.from);
        }

        @Override
        public String toString() {
            return "PrepareKey{" + "op=" + op + ", from=" + from + '}';
        }
        
    }
    
    private static class SeatKey {
        private final FlightId flight;
        private final CustomerId customer;
        private final ServerId from;

        public SeatKey(FlightId flight, CustomerId customer, ServerId from) {
            this.flight = flight;
            this.customer = customer;
            this.from = from;
        }

        @Override
        public int hashCode() {
            int hash = 7;
            hash = 97 * hash + Objects.hashCode(this.flight);
            hash = 97 * hash + Objects.hashCode(this.customer);
            hash = 97 * hash + Objects.hashCode(this.from);
            return hash;
        }

        @Override
        public boolean equals(Object obj) {
            if (obj == null) {
                return false;
            }
            if (getClass() != obj.getClass()) {
                return false;
            }
            final SeatKey other = (SeatKey) obj;
            if (!Objects.equals(this.flight, other.flight)) {
                return false;
            }
            if (!Objects.equals(this.customer, other.customer)) {
                return false;
            }
            return Objects.equals(this.from, other.from);
        }

        @Override
        public String toString() {
            return "SeatKey{" + "flight=" + flight + ", customer=" + customer + ", from=" + from + '}';
        }
        
        
    }
}
